package com.zhuwei.kafka.producer;

import com.zhuwei.kafka.constant.Topic;
import org.apache.kafka.clients.producer.*;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

/**
 * 功能：生产者，2.11-0.11.0.2之后的新API
 * @author zhuiwei
 * @version 2019/05/14
 */
public class NewProducer {

    public static void main(String[] args) {

        Properties props = new Properties();
        // Kafka服务端的主机名和端口号
        props.put("bootstrap.servers", "192.168.154.101:9092");
        // 等待所有副本节点的应答
        props.put("acks", "all");
        // 消息发送最大尝试次数
        props.put("retries", 0);
        // 一批消息处理大小
        props.put("batch.size", 16384);
        // 请求延时
        props.put("linger.ms", 1);
        // 发送缓存区内存大小
        props.put("buffer.memory", 33554432);
        // key序列化
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        // value序列化
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        // 2 构建拦截链
        List<String> interceptors = new ArrayList<>();
        interceptors.add("com.zhuwei.kafka.interceptor.TimeInterceptor");
        interceptors.add("com.zhuwei.kafka.interceptor.CounterInterceptor");
        props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, interceptors);

        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
        //发送消息，回调处理
//        String data="hello world";
//        producer.send(new ProducerRecord<String, String>(Topic.FIRST, data)
//                , new Callback() {
//                    @Override
//                    public void onCompletion(RecordMetadata metadata, Exception exception) {
//                        if (metadata != null) { ;
//                            System.out.println(metadata.partition() + "---" + metadata.offset());
//                        }
//                    }
//                }
//        );
        //此处为interceptor案例的片段测试代码
        for (int i = 0; i < 5; i++) {
            producer.send(new ProducerRecord<>(Topic.FIRST, "好" + i));
        }

        //若要调用interceptor的close方法，处一定要关闭
        producer.close();
    }
}
